package com.hba.consumer;

import com.hba.util.HessionCodecFactory;
import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.core.ChannelAwareMessageListener;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.amqp.support.converter.SimpleMessageConverter;
import org.springframework.context.support.ApplicationObjectSupport;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

/**
 * 消息接收方
 */
@Component
public class MessageReceiver extends ApplicationObjectSupport implements ChannelAwareMessageListener {

	private static final Logger logger = LoggerFactory.getLogger(MessageReceiver.class);

	public	void onMessage(Message message, Channel channel) throws Exception {
		MessageBody messageBody = null;
		try{
			MessageConverter messageConverter = new SimpleMessageConverter();
			Object messageObject = messageConverter.fromMessage(message);
			if (!(messageObject instanceof MessageBody)){
				logger.error("receive message not instanceof MessageBody, message=" + messageObject);
			}
			messageBody=(MessageBody) messageObject;
			logger.debug("receive {}",messageBody);
			StringBuilder queueBuild = new StringBuilder();
			queueBuild.append(messageBody.getExchangeName());
			queueBuild.append("|");
			queueBuild.append(messageBody.getQueueName());

			Object obj = HessionCodecFactory.deSerialize(messageBody.getMessageData());
			String beanName = MessageConsumer.consumerMap.get(queueBuild.toString());
			if (StringUtils.isEmpty(beanName)){
				logger.error("receive bean is null! queueBuild=" + queueBuild.toString()
						+ ", message=" + messageBody);
			}
			IMessageProcess messageProcess =(IMessageProcess) this.getApplicationContext().getBean(beanName);
			messageProcess.process(obj);
			channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
		}catch(Exception ex){
			logger.error("receive message body:" + messageBody);
			logger.error("receive error :", ex);
			channel.basicNack(message.getMessageProperties().getDeliveryTag(), false,false);
		}
	}
}
